package com.wq.sofa.framework.bolt.serialize;

import com.alipay.remoting.DefaultCustomSerializer;
import com.alipay.remoting.InvokeContext;
import com.alipay.remoting.exception.DeserializationException;
import com.alipay.remoting.exception.SerializationException;
import com.alipay.remoting.rpc.RequestCommand;
import com.alipay.remoting.rpc.protocol.RpcRequestCommand;
import com.wq.sofa.framework.bolt.model.CommonRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @Author: wangqiang20995
 * @Date:2018/9/29
 * @Description:
 * @Resource:
 */
public class DefaultJdkSerializer extends DefaultCustomSerializer {

    private Logger logger = LoggerFactory.getLogger(getClass());

    private static final String EXCEPTION_SERIALIZER_MESSAGE = "serialize exception";

    private static final String WRONG_CLASS_SERIALIZE = "wrong class";

    private Map<Class,AtomicBoolean> communicationStore ;

    //not thread safe
    private ByteArrayOutputStream byteArrayOutputStream;
    private ByteArrayInputStream byteArrayInputStream;


    public DefaultJdkSerializer(){
        communicationStore = new ConcurrentHashMap<>(1000);
        byteArrayOutputStream = new ByteArrayOutputStream();
    }


    //object --> byte[]
    @Override
    public <T extends RequestCommand> boolean serializeContent(T request, InvokeContext invokeContext) throws SerializationException {
        if(request instanceof RpcRequestCommand){
            logger.info("custom serializer do serial...");
            CommonRequest commonRequest = (CommonRequest) ((RpcRequestCommand) request).getRequestObject();
            byte[] data;
            ObjectOutputStream objectOutputStream = null;
            try {
                objectOutputStream = new ObjectOutputStream(this.byteArrayOutputStream);
                objectOutputStream.writeObject(commonRequest);
                objectOutputStream.flush();
                data = this.byteArrayOutputStream.toByteArray();
            } catch (IOException e) {
                logger.error(e.getMessage(),e);
                data = EXCEPTION_SERIALIZER_MESSAGE.getBytes();
            }finally {
                if(objectOutputStream != null){
                    try {
                        objectOutputStream.close();
                        this.byteArrayOutputStream.close();
                    } catch (IOException e) {
                        logger.error(e.getMessage(),e);
                    }
                }

            }

            request.setContent(data);
            return true;
        }

        throw new SerializationException("wrong serial type,expect is ["+RpcRequestCommand.class.getName()+"] but serial type is ["+request.getClass().getName()+"]");
    }


    @Override
    public <T extends RequestCommand> boolean deserializeContent(T request) throws DeserializationException {
        RpcRequestCommand requestCommand = (RpcRequestCommand) request;
        byte[] buffer = request.getContent();
        ObjectInputStream objectInputStream = null;
        logger.info("custom serializer to deSerial...");
        try{

            this.byteArrayInputStream = new ByteArrayInputStream(buffer);

            objectInputStream = new ObjectInputStream(this.byteArrayInputStream);
            Object o = objectInputStream.readObject();
            if(o instanceof CommonRequest){
                requestCommand.setRequestObject(o);
            }else {
                requestCommand.setRequestObject(WRONG_CLASS_SERIALIZE);
            }

        }catch (IOException e){
            logger.error(e.getMessage(),e);
        } catch (ClassNotFoundException e) {
            logger.error(e.getMessage(),e);
        }finally {
            if(objectInputStream != null){
                try {
                    objectInputStream.close();
                    this.byteArrayInputStream.close();
                } catch (IOException e) {
                    logger.error(e.getMessage(),e);
                }
            }
        }

        return true;
    }
}
